﻿using Basic.EventBus.Eventbus;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;

namespace Basic.EventBus.EventBusRabbitMQ
{
    internal class RabbitMQSubscriber : IEventSubscriber
    {
        private static readonly string ExchangeType = "topic";

        private ILogger<RabbitMQSubscriber> Logger;
        private List<RabbitConsumerRecord> SubscriberInfos;
        private RabbitMQConnections Connections;

        public RabbitMQSubscriber(ILogger<RabbitMQSubscriber> logger, RabbitMQConnections connections)
        {
            Logger = logger;
            Connections = connections;
            SubscriberInfos = new List<RabbitConsumerRecord>(16);
        }

        public void AddSubscriber(ISubscriberHandler subscriberInfo)
        {
            var conn = Connections.GetConnection();
            try
            {
                if (!conn.IsOpen) throw new ConnectFailureException("IConnection.IsOpen is false;", null);

                var channel = conn.CreateModel();
                if (!channel.IsOpen) throw new ConnectFailureException("IModel.IsOpen is false;", null);

                bool autoAck = Connections.BrokerConfig.AutoAck;
                ushort qos = Connections.BrokerConfig.Qos;
                string exchangeName = subscriberInfo.Attribute.Group;
                string queueName = subscriberInfo.Attribute.Topic;
                bool durable = subscriberInfo.Attribute.Durable;

                channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType, true, autoDelete: false, null);
                channel.QueueDeclare(queue: queueName, durable, exclusive: false, autoDelete: !durable, null);
                channel.QueueBind(queueName, exchangeName, subscriberInfo.Attribute.Key, null);

                if (qos > 0) channel.BasicQos(0, qos, false);

                RabbitConsumerRecord consumerInfo = new RabbitConsumerRecord(channel, subscriberInfo, Connections.BrokerConfig, Logger);
                consumerInfo.Start();
                SubscriberInfos.Add(consumerInfo);
            }
            catch
            {
                throw;
            }
            finally
            {
                Connections.Return(conn);
            }
        }

        public void AddSubscriber(ISubscriberHandler[] subscriberInfos)
        {
            if (subscriberInfos?.Length > 0)
                foreach (var item in subscriberInfos)
                    AddSubscriber(item);
        }

        #region Dispose
        private bool disposedValue;

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    // TODO: 释放托管状态(托管对象)
                }
                // TODO: 将大型字段设置为 null
                disposedValue = true;
            }
        }

        public void Dispose()
        {
            // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
        #endregion

        #region 保存消费者信息的内部类

        internal class RabbitConsumerRecord
        {
            private ILogger Logger;
            private bool AutoAck;
            private ushort Qos;
            private SemaphoreSlim ConcurrentLock;
            private EventingBasicConsumer Consumer;
            RabbitMQBusBrokerConfig Config;

            public ISubscriberHandler Subscriber { get; private set; }
            public IModel Channel { get; private set; }

            //public string ConsumerTag { get; private set; }
            public DateTime CreateTime { get; private set; }


            public RabbitConsumerRecord(IModel channel, ISubscriberHandler subscriber, RabbitMQBusBrokerConfig config, ILogger logger)
            {
                Logger = logger;
                Config = config;
                Channel = channel;
                Subscriber = subscriber;
                AutoAck = config.AutoAck;
                Qos = config.Qos;
                if (config.Qos > 0)
                {
                    int initNum = 8;
                    if (initNum > config.Qos) initNum = config.Qos;
                    ConcurrentLock = new SemaphoreSlim(initNum, config.Qos);
                }
                CreateTime = DateTime.Now;
            }

            private void CreateConsumer(ISubscriberHandler subscriberInfo)
            {
                Consumer = new EventingBasicConsumer(Channel);
                Action<object, BasicDeliverEventArgs> handler = async (sender, args) =>
                {
                    try
                    {
                        byte[] data = args.Body.Span.ToArray();

                        if (ConcurrentLock != null)
                            await ConcurrentLock.WaitAsync();
                        ThreadPool.QueueUserWorkItem(async obj =>
                        {
                            await subscriberInfo.HandleMessage(obj as byte[]);
                            ConcurrentLock?.Release();
                        }, data);

                        if (!AutoAck) Channel.BasicAck(args.DeliveryTag, false);
                    }
                    catch (Exception ex)
                    {
                        Logger.LogError(ex, $"ISubscriberHandler.HandleMessage(messageType:{subscriberInfo.MessageType.FullName}) Error:");
                    }
                };

                Consumer.Received += new EventHandler<BasicDeliverEventArgs>(handler);
            }

            /// <summary>
            /// 启动
            /// </summary>
            public void Start()
            {
                if (Config.PushModel)
                {
                    CreateConsumer(Subscriber);
                    Channel.BasicConsume(Subscriber.Attribute.Topic, AutoAck, Consumer);
                }
                else
                {
                    _ = Task.Factory.StartNew(PullReceived, Subscriber.TokenSource.Token);
                }
            }

            private async Task PullReceived()
            {
                string topic = Subscriber.Attribute.Topic;
                bool autoAck = AutoAck;
                while (Subscriber.TokenSource.Token.CanBeCanceled)
                {
                    try
                    {
                        if (ConcurrentLock != null)
                            await ConcurrentLock.WaitAsync();
                        var result = Channel.BasicGet(topic, autoAck);
                        byte[] data = result.Body.Span.ToArray();
                        ThreadPool.QueueUserWorkItem(async obj =>
                        {
                            await Subscriber.HandleMessage(obj as byte[]);
                            ConcurrentLock?.Release();
                        }, data);

                        if (!AutoAck) Channel.BasicAck(result.DeliveryTag, false);
                        Channel.BasicReject(result.DeliveryTag, false);
                    }
                    catch (Exception ex)
                    {
                        Logger.LogError(ex, $"ISubscriberHandler.HandleMessage(messageType:{Subscriber.Attribute.Topic}) Error:");
                    }
                }
            }

        }

        #endregion

    }
}